import json
import time

from absl import app, flags

from concurrent.futures import ProcessPoolExecutor, as_completed
from cc.experimental.wliang.ksqldb.ksqldb_client import KSqldbClient
from xunkemgmt_client.tool.slack_noti import send_to_slack


FLAGS = flags.FLAGS


def notice_slack(report, slack, subject, context):
  if slack is None:
    return
  send_to_slack(context, slack, 'msg', subject)


def _process_order_event(rows, kafka_topic):
  fetched_time = 0
  header = next(rows)
  for row in rows:
    now = time.time_ns()
    if fetched_time + 30000000000 > now:
      continue
    fetched_time = now
    j = row.replace(",", "")
    info = json.loads(j)
    slack = "@wliang" #"hft_dev_order"
    context = f"%s got %s fail orders in 5 minutes" % (kafka_topic, info['row']['columns'][0])
    subject = f"fail order notice"
    notice_slack(None, slack, subject, context)


def _monitor_order_event(ksql_client):
  # create base stream
  base_stream = ksql_client.initial()
  if base_stream is None:
    # maybe no topic
    return

  # create order event stream
  create_stream = "CREATE STREAM {}_ORDER_EVENT AS SELECT og_log FROM {} WHERE type = 'OG_LOG' AND og_log->type = 'ORDER_EVENT' EMIT CHANGES;".format(base_stream, base_stream)
  if not ksql_client.execute_no_throw(create_stream):
    return

  # monitor stream
  # query_stream = "SELECT og_log FROM {}_ORDER_EVENT EMIT CHANGES;".format(base_stream)
  # ksql_client.consume(query_stream, _process_order_event)

  # create order rejected stream
  create_stream = "CREATE OR REPLACE STREAM {}_ORDER_REJECTED AS SELECT og_log->event->type AS type, og_log->event As event FROM {}_ORDER_EVENT WHERE og_log->account_request->exchange != 'Ftx' AND og_log->event->type = 'ORDER_REJECTED' EMIT CHANGES;".format(base_stream, base_stream)
  if not ksql_client.execute_no_throw(create_stream):
    return

  # monitor stream
  # query_stream = "SELECT * FROM {}_ORDER_REJECTED EMIT CHANGES;".format(base_stream)
  # ksql_client.consume(query_stream, _process_order_event)

  # create order rejected in 5 minnutes table
  create_table = "CREATE TABLE {}_ORDER_REJECTED_PER_5MINS AS SELECT type AS type, COUNT(type) AS event_cnt FROM {}_ORDER_REJECTED WINDOW TUMBLING (SIZE 5 MINUTE) GROUP BY type EMIT CHANGES;".format(base_stream, base_stream)
  if not ksql_client.execute_no_throw(create_table):
    return

  # monitor order rejected in 5 minutes
  ksql_query= "SELECT event_cnt FROM {}_ORDER_REJECTED_PER_5MINS WHERE event_cnt > 50 EMIT CHANGES;".format(base_stream)
  ksql_client.consume(ksql_query, _process_order_event)


def _consume(kafka_topic, ksqldb_server):
  ksql_client = KSqldbClient(ksqldb_server, kafka_topic)

  while True:
    try:
      _monitor_order_event(ksql_client)
      time.sleep(5)
    except Exception as e:
      import traceback
      print(traceback.format_exc())
      print("%s error. retry after 5s ..." % FLAGS.kafka_topic)
      time.sleep(5)


def main(_):
  assert(FLAGS.kafka_topics)
  ksqldb_server = FLAGS.ksqldb_server
  kafka_topics = FLAGS.kafka_topics.split(',')
  futures = []
  with ProcessPoolExecutor(max_workers=len(kafka_topics)) as executor:
    for kafka_topic in kafka_topics:
      futures.append(executor.submit(_consume, kafka_topic, ksqldb_server))
      time.sleep(5)
    for fut in as_completed(futures):
      _ = fut.result()


if __name__ == '__main__':
  flags.DEFINE_string('ksqldb_server', "http://localhost:8088", "ksqldb_server")
  flags.DEFINE_string('kafka_topics', None, "kafka topics")
  app.run(main)
